import backtrader as bt
import talib.deprecated
import talib as ta
import pandas as pd
import numpy as np
import math
import time
import logging
import datetime
from collections import namedtuple
from multiprocessing import Process, Pipe, Queue
from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
from apscheduler.jobstores.memory import MemoryJobStore
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.schedulers.blocking import BlockingScheduler
from pyLibs import GadflyUnits as units
from pyLibs.Gadfly import StockStore, BackTraderStrategy as bts, BackTraderIndicator as bti, BackTraderUtits as btu

logging.basicConfig()
logging.getLogger('apscheduler').setLevel(logging.DEBUG)


class TestStrategy(bts.SingleCycleStrategy):
    def __init__(self, config):
        super().__init__()
        self.config = config.Strategy

    def next(self):
        trend = list()
        print('开始分析时间：%s' % datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3])
        try:
            for i, data in enumerate(self.datas):
                inds = self.indicators[i]
                cond = self.condition[i]
                if str(data.datetime.date(0)) == datetime.datetime.now().strftime('%Y-%m-%d'):
                    print(i, data.datetime.datetime(0), data._name, data.close[0])
                    '''if i % 2 == 1:
                        if cond['is_vol_release'][0] and data.volume[0] >= data.volume[-1] * 2 \
                                and cond['close_cross_ema_s'][0]:
                            pp = (data.datetime.datetime(0), data._name, data.close[0], inds['middle'][0], inds['upper'][0],
                                  inds['lower'][0], inds['ema_s'][0], inds['ema_m'][0])
                            print('%s：%s，收盘价：%.3f，中轨：%.3f，上轨：%.3f，下轨：%.3f，EMA13：%.3f，EMA34：%.3f，满足条件！' % pp)
                            trend.append(True)
                        elif data.close[0] > inds['ema_m'][0] and inds['middle'][0] > inds['ema_m'][0]:
                            pp = (data.datetime.datetime(0), data._name, data.close[0], inds['middle'][0], inds['upper'][0],
                                  inds['lower'][0], inds['ema_s'][0], inds['ema_m'][0], inds['bb'][0], inds['width'][0],
                                  inds['atr_loss'][0], inds['platform_upper_high'][0], inds['platform_lower_low'][0])
                            print('%s：%s，收盘价：%.3f，中轨：%.3f，上轨：%.3f，下轨：%.3f，EMA13：%.3f，EMA34：%.3f，BB：%.3f，'
                                  'Width：%.3f，ATR止损价：%.3f，平台上沿：%.3f，平台下沿：%.3f' % pp)
                            trend.append(True)
                        else:
                            trend.append(False)
                    else:
                        if data.close[0] >= inds['middle'][0]:
                            pp = (data.datetime.datetime(0), data._name, data.close[0], inds['middle'][0], inds['upper'][0],
                                  inds['lower'][0], inds['ema_s'][0], inds['ema_m'][0], inds['bb'][0], inds['width'][0],
                                  inds['atr_loss'][0], inds['platform_upper_high'][0], inds['platform_lower_low'][0])
                            print('%s：%s，收盘价：%.3f，中轨：%.3f，上轨：%.3f，下轨：%.3f，EMA13：%.3f，EMA34：%.3f，BB：%.3f，'
                                  'Width：%.3f，ATR止损价：%.3f，平台上沿：%.3f，平台下沿：%.3f' % pp)'''
        except Exception as e:
            print(e)
        finally:
            print('结束分析时间：%s' % datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3])



def cerebroJob(stock_list, queue, pipe_c, config):
    cerebro = bt.Cerebro()
    cerebro.addstrategy(TestStrategy, config=config)
    tdxStore = StockStore.TdxStore(config)
    history = tdxStore.liveResampleDatas(stock_list, config.Scheduler.store.bars)
    for key, d in history.items():
        # todo: ddd
        cerebro.adddata(btu.StockLiveData(queue=queue[key], pipe=pipe_c, dataname=d.data,
                                          name=key, timeframe=d.cycle.timeframe,
                                          compression=d.cycle.compression,
                                          fromdate=config.Scheduler.store.fromdate,
                                          todate=config.Scheduler.store.todate))
    cerebro.run()


def jobExecutes(stocks, queue, config):
    tdx = StockStore.TdxStore(config)
    live_dp = tdx.getStockBarsFromApi(stocks, 0, 2, True)
    for dp in live_dp:
        for row in dp.data.itertuples(index=True, name='Stock'):
            rows = row._asdict()
            rows.pop('Index')
            live = pd.DataFrame(rows, index=[0])
            live.index = pd.to_datetime(live.datetime)
            queue[dp.info.tdx_code + '_' + dp.cycle.cycle].put(live)


def schedulerJob(stocks, queue, pipe, config, start_time):
    job = {
        'default': {'coalesce': False, 'max_instances': 3},
        'store': {'default': MemoryJobStore()},
    }
    if config.Scheduler.pool.capitalize() == 'Thread':  # 线程池执行器
        job['executor'] = {'default': ThreadPoolExecutor(config.Scheduler.pool_num)}
    elif config.Scheduler.pool.capitalize() == 'Process':  # 进程池执行器
        job['executor'] = {'default': ProcessPoolExecutor(config.Scheduler.pool_num)}
    if config.Scheduler.executor_type.capitalize() == 'Background':
        scheduler = BackgroundScheduler(executors=job['executor'], job_defaults=job['default'],
                                        jobstores=job['store'], timezone='Asia/Shanghai')
    else:
        scheduler = BlockingScheduler(executors=job['executor'], job_defaults=job['default'],
                                      jobstores=job['store'], timezone='Asia/Shanghai')
    scheduler.add_job(jobExecutes, 'cron', day_of_week='mon-fri', hour='9', minute='45',
                      second='3', args=[stocks, queue, config])
    scheduler.add_job(jobExecutes, 'cron', day_of_week='mon-fri', hour='10, 13, 14', minute='00, 30, 45',
                      second='3', args=[stocks, queue, config])
    scheduler.add_job(jobExecutes, 'cron', day_of_week='mon-fri', hour='11', minute='00, 15, 30',
                      second='3', args=[stocks, queue, config])
    scheduler.add_job(jobExecutes, 'cron', day_of_week='mon-fri', hour='9-23', minute='00, 15, 30, 45', second='3',
                      args=[stocks, queue, config])
    historical_data_load_success = list()
    while True:
        message = pipe.recv()
        historical_data_load_success.append(message)
        print(str(len(historical_data_load_success)) + '.' + message)
        if len(historical_data_load_success) == len(stocks):
            run_time_ms = int(round(time.time() * 1000)) - start_time
            seconds, milliseconds = divmod(run_time_ms, 1000)
            minutes, seconds = divmod(seconds, 60)
            hours, minutes = divmod(minutes, 60)
            run_time = "%d:%d:%d" % (hours, minutes, seconds)
            dt = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
            print('%s，所有历史数据均已加载成功，系统即将进入实盘状态！加载历史数据总耗时：%s' % (dt, run_time))
            scheduler.start()
        time.sleep(1)


def cerebroSchedulerMain():
    start_time = int(round(time.time() * 1000))
    print('%s，系统开始运行！' % datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3])
    config = units.loadYaml('./BackTrader/config.yaml')
    (pipe_c, pipe_s) = Pipe()
    #stocks = ['600031', '000561', '159949']
    stocks = StockStore.get_stock_block(['沪深300', '创业50'], 'zs', config.Source.tdx.win_path, False)
    cycles = ['15T', '60T']
    stock_list = list()
    queue = dict()
    for s in stocks:
        for c in cycles:
            rule = StockStore.get_stock_cycle(c)
            info = StockStore.get_stock_market(s)
            queue[info.tdx_code + '_' + rule.cycle] = Queue()
            stock_list.append((s, c))
    cerebro_process = Process(target=cerebroJob, args=(stock_list, queue, pipe_c, config,))  # 创建一个 Cerebro 进程
    scheduler_process = Process(target=schedulerJob, args=(stock_list, queue, pipe_s, config, start_time,))  # 创建一个 Scheduler 进程
    cerebro_process.start()
    scheduler_process.start()


if __name__ == '__main__':
    cerebroSchedulerMain()
